package com.ebupt.migu.music.autoTri.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;

/**
 * @author xuhongsong
 * @version 1.0
 * @date 2023/4/10 9:39
 */
@Service
@Slf4j
public class KafkaSender {


    @Resource
    private KafkaTemplate kafkaTemplate;

   public void pro_test(String topic,String jsonString){
       log.info("==============top:{},jsonString:{}",topic,jsonString);
      /* ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(new ProducerRecord<>(topic,
               jsonString));*/
      // kafkaTemplate.send(topic,jsonString);

       kafkaTemplate.send(topic, jsonString).addCallback(success -> {
           // 消息发送到的topic
      //     String topic = success.getRecordMetadata().topic();
        /*   // 消息发送到的分区
           int partition = success.getRecordMetadata().partition();
           // 消息在分区内的offset
           long offset = success.getRecordMetadata().offset();*/
          log.info("发送消息成功: topic:{},json体：{}" ,topic,jsonString);
       }, failure -> {
           log.error("发送消息失败:" + failure.getMessage());
       });
    }

}
